
#include "../iface/debug.h"
#include "../iface/APRInterface.h"
#include "../utils/StringUtils.h"
#include "APRAutoLockRead.h"
#include "APRAutoLockWrite.h"
#include "APRBucketBrigade.h"

/** Construct a new bucket brigade class. Using a memory pool (a non-NULL
 * argument) is recommended.
 * 
 * @param imp
 * The APR memory pool to be parent to the new APR memory pool of this
 * instance. Can be NULL. */
APRBucketBrigade::APRBucketBrigade(apr_pool_t *imp)
{
   apr_status_t status;
   status = apr_pool_create(&mp, imp);
   CheckAPRError(status);
   parent_mp = imp;
   bucketalloc = apr_bucket_alloc_create(mp);
   bucketbrgde = apr_brigade_create(mp, bucketalloc);
   rwlock = new APRRWMutex(mp);
   csize = 0; // csize is amount in memory, not real length
}

/** Destructor for the bucket brigade class. */
APRBucketBrigade::~APRBucketBrigade()
{
   apr_status_t status;
   status = apr_brigade_destroy(bucketbrgde);
   CheckAPRError(status);
   apr_bucket_alloc_destroy(bucketalloc);
   delete rwlock;
   apr_pool_destroy(mp);
}

/** Enables adding an open socket to the bucket by reference. Through this
 * socket data can be streamed by reading from the bucket.
 * 
 * @param tsock
 * An open socket. */
void APRBucketBrigade::addSocketBucket(apr_socket_t *tsock)
{
   APRAutoLockWrite awlock(rwlock);
   apr_bucket *tbucket = apr_bucket_socket_create(tsock, bucketalloc);
   APR_BRIGADE_INSERT_TAIL(bucketbrgde, tbucket);
}

/** Write end-of-stream marker to the bucket brigade indicating an end of the
 * data stream. */
void APRBucketBrigade::writeEOS()
{
   APRAutoLockWrite awlock(rwlock);
   apr_bucket *tbucket = apr_bucket_eos_create(bucketalloc);
   APR_BRIGADE_INSERT_TAIL(bucketbrgde, tbucket);
}

/** Enables adding a file, or a byte range from a file, by reference into the
 * bucket brigade stream. The contents of the file can now be accessed by
 * reading the stream. The size of the file, or the number of bytes believed
 * to have been added to the stream is returned.
 * 
 * @param filename
 * The name of the file to be added to the stream.
 * 
 * @param start
 * The start of the byte range from the file to be added to the stream.
 * 
 * @param end
 * The end of the byte range from the file to be added to the stream.
 * 
 * @param isbinary
 * Does the binary flag need to be appended to the file reading flags.
 * 
 * @return
 * Number of bytes estimated to have been successfully added to the stream. */
apr_size_t APRBucketBrigade::writeFileToStream(const char *filename, apr_size_t start, apr_size_t end, int isbinary)
{
   apr_status_t status;
   apr_size_t eend = end;
   apr_file_t *newFP;
   status = apr_file_open(&newFP, filename, ((isbinary) ? APR_FOPEN_BINARY : 0)|APR_FOPEN_READ|APR_FOPEN_BUFFERED, APR_FPROT_OS_DEFAULT, mp);
   CheckAPRError(status);
   if (eend == 0)
   {
      apr_finfo_t finfo;
#ifdef USE_STAT_FILE_SIZE
      status = apr_file_info_get(&finfo, APR_FINFO_SIZE, newFP);
      CheckAPRError(status);
#else
      apr_off_t offset = 0;
      status = apr_file_seek(newFP, APR_END, &offset);
      CheckAPRError(status);
      finfo.size = offset;
      offset = 0;
      status = apr_file_seek(newFP, APR_SET, &offset);
      CheckAPRError(status);
#endif
      eend = finfo.size;
   }

   apr_bucket *fileb = apr_bucket_file_create(newFP, start, eend - start, mp, bucketalloc);
   if (fileb == NULL)
      errprint("bucket allocation failed");
   APRAutoLockWrite awlock(rwlock);
   APR_BRIGADE_INSERT_TAIL(bucketbrgde, fileb);
   csize += fileb->length;
   return (eend - start);
}

/** Write text data via a format string into the bucket brigade stream. The
 * number of bytes added to the stream is returned.
 * 
 * @param format
 * Format string in the style of printf.
 * 
 * @return
 * Number of bytes written to the stream. */
apr_size_t APRBucketBrigade::writefToStream(const char *format, ...)
{
   va_list args;
   va_start(args, format);
   char *actual = StringUtils::varprintf(format, args);
   va_end(args);
   apr_size_t result = writeToStream(actual, strlen(actual));
   free(actual);
   return result;
}

/** Write data from a buffer into the bucket brigade stream. The number of
 * bytes added to the stream is returned
 * 
 * @param data
 * Input data buffer.
 * 
 * @param numbytes
 * Number of bytes of data from the buffer to suffix to the stream.
 * 
 * @return
 * Number of bytes written to the stream. */
apr_size_t APRBucketBrigade::writeToStream(const char *data, apr_size_t numbytes)
{
   if (numbytes == 0) return 0;
   char *bbuf = (char *)apr_bucket_alloc(numbytes, bucketalloc);
   if (!bbuf) return 0;
   memcpy(bbuf, data, numbytes);
   APRAutoLockWrite awlock(rwlock);
   apr_bucket *tbucket = apr_bucket_heap_create(bbuf, numbytes, apr_bucket_free, bucketalloc);
   APR_BRIGADE_INSERT_TAIL(bucketbrgde, tbucket);
   csize += numbytes;
   return numbytes;
}

/** Read data from the bucket into a buffer. The number of bytes read from the
 * stream is returned.
 * 
 * @param[out] data
 * Output data buffer.
 * 
 * @param numbytes
 * Number of bytes of data from the stream to put into the buffer.
 * 
 * @return
 * Number of bytes read from the stream. */
apr_size_t APRBucketBrigade::readFromStream(char *data, apr_size_t numbytes)
{
   APRAutoLockRead arlock(rwlock);
   apr_status_t status;
   apr_size_t actual_length = numbytes;
   status = apr_brigade_flatten(bucketbrgde, data, &actual_length);
   CheckAPRError(status);
   return actual_length;
}

/** Delete data from the bucket brigade. The number of bytes actually
 * deleted from the stream is returned.
 * 
 * @param numbytes
 * Number of bytes to remove from the beginning of the stream.
 * 
 * @return
 * Number of bytes actually removed form the beginning of the stream. */
apr_size_t APRBucketBrigade::removeFromStream(apr_size_t numbytes)
{
   APRAutoLockWrite awlock(rwlock);
   apr_size_t remaining = numbytes;
   apr_bucket *tbucket = APR_BRIGADE_FIRST(bucketbrgde);
   apr_bucket *tnext = NULL;
   while (tbucket != APR_BRIGADE_SENTINEL(bucketbrgde))
   {
      if (APR_BUCKET_IS_EOS(tbucket)) break;
      if (APR_BUCKET_IS_SOCKET(tbucket)) { remaining = 0; break; }
      tnext = APR_BUCKET_NEXT(tbucket);
      if (tbucket->length > remaining)
      {
         tbucket->start += remaining;
         tbucket->length -= remaining;
         csize -= remaining;
         remaining = 0;
         break;
      } else
      {
         remaining -= tbucket->length;
         csize -= tbucket->length;
         apr_bucket_delete(tbucket);
      }
      tbucket = tnext;
   }
   return numbytes - remaining;
}

/** Pop data (read and delete data) from the bucket brigade. The number of
 * bytes obtained is returned.
 * 
 * @param[out] data
 * Place to put the data once read.
 * 
 * @param numbytes
 * Number of bytes to read and remove (pop).
 * 
 * @return
 * Number of bytes removed from stream (always less than or equal to the 
 * number of bytes read). */
apr_size_t APRBucketBrigade::popFromStream(char *data, int numbytes)
{
   apr_size_t read = readFromStream(data, numbytes);
   apr_size_t length = realLength();
   apr_size_t pop = removeFromStream(read);
   if (pop != read)
      errprint("Read and deleted data do not match, read %d bytes, deleted %d bytes (length reported to be %d).", read, pop, length);
   return pop;
}

/** Put chunks of data into the bucket brigade. Returns zero when successful,
 * less than zero otherwise.
 * 
 * @param buffer
 * Big buffer from which to take data and suffix it onto the stream represented
 * by this bucket brigade.
 * 
 * @param[in,out] length
 * The intended number of bytes from the big buffer to suffix to the stream.
 * On return, this is the number of bytes actually added to the stream.
 * 
 * @param smallblock
 * This is the block size of the copy which copies in chunks to prevent the
 * allocation of a huge continuous bucket that could overload buffers
 * elsewhere.
 * 
 * @return
 * Zero on success, non-zero on error. */
int APRBucketBrigade::bufferToBucketBrigade(const char *buffer, size_t *length, size_t smallblock)
{
   for (int k = 0; k < (*length); k += smallblock)
   {
      int actualamt = (((*length) - k) < smallblock) ? (*length) - k : smallblock;
      int bytes = writeToStream(&(buffer[k]), actualamt);
      if (bytes == 0)
      {
         errprint("push buffer to bucket brigade failed - incomplete data (%d/%d bytes written)", k, (*length));
         if (length) *length = k;
         writeEOS();
         return -1;
      }
      k -= (smallblock - bytes);
   }
   writeEOS();
   return 0;
}

/** Read data from the bucket brigade into a buffer until the end of the data
 * stream. Returns zero when successful, less than zero otherwise.
 * 
 * @param[out] buffer
 * A pointer to the place where a pointer to the new buffer is to be placed.
 * The responsibility for freeing the buffer rests with the caller.
 * 
 * @param[out] length
 * The number of bytes in the buffer when the bucket brigade has been fully
 * read.
 * 
 * @return
 * Zero on success, non-zero on failure. */
int APRBucketBrigade::bucketBrigadeToBuffer(char **buffer, size_t *length)
{
   apr_size_t reallen = realLength();
   (*buffer) = (char *)malloc(sizeof(char)*(reallen+1));
   if (reallen > 0)
   {
      apr_size_t gotreal = popFromStream((*buffer), reallen);
      if (gotreal != reallen)
      {
         errprint("Could not get all of the bucket brigade (%" APR_SIZE_T_FMT "/%" APR_SIZE_T_FMT " bytes obtained)", gotreal, reallen);
         if (length) (*length) = gotreal;
         return -1;
      }
   }
   if (length) (*length) = reallen;
   return 0;
}
